package com.hna.eking.ZkUtils;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZkSeqLock implements Watcher{
	private int SESSION_TIMEOUT = 50000; 
	protected String hosts = "127.0.0.1:2181"; 
	private String FilePath = "/Lock";
	private ZooKeeper zk;
	private Condition cond;
	private Lock lock;
	private String MyNode ;
	public final static String PLT_STATUS_RUN  = "0001";
	public final static String PLT_STATUS_SUC  = "0002";
	public final static String PLT_STATUS_FAIL  = "0003";
	
	private static final Logger logger = LoggerFactory.getLogger(ZkSeqLock.class);

	@Override
	public void process(WatchedEvent evt) {
		// TODO Auto-generated method stub
		logger.debug("get watcher:" + evt.getType());
		
		if(evt.getType() !=  Watcher.Event.EventType.NodeChildrenChanged){
			logger.debug("get watcher===>:" + evt.getType());
			
			return ;
		}
		UnLock();
		logger.debug("process watcher: Unlock");
	}
	
	public void InitWatcher(){
		try {
			zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
			
			logger.debug("file:" + FilePath);
			Stat st = zk.exists(FilePath, false);
			logger.debug("file run here ");
			if(st == null){
				logger.debug("file " + FilePath + " not exists");
				zk.create(FilePath, PLT_STATUS_RUN.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); 
			}
//			MyNode = zk.create(FilePath, PLT_STATUS_RUN.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); 
//			zk.create(FilePath + "/SEQ", PLT_STATUS_RUN.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); 
			MyNode = zk.create(FilePath + "/SEQ", PLT_STATUS_RUN.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); 
			lock = new ReentrantLock();
			cond = lock.newCondition();
		} catch (IOException | KeeperException | InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	
	public void WaitLock(){
		lock.lock();
		logger.debug("Condition locked!");
		try {
			List<String>ChildList = zk.getChildren(FilePath, this);
			
//			for(String child : ChildList){
//				logger.debug("child file is:" + child);
//			}
			//zk.getData(FilePath, this, new Stat());
			Collections.sort(ChildList);
			if(!MyNode.equals(ChildList.get(0))){
				logger.debug("Condition await! " + MyNode);
				cond.await();
			}
			
			
		} catch (InterruptedException | KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
			logger.debug("Condition unlock!");
			lock.unlock();
		}
		
		
	}
	
	public void UnLock(){
		lock.lock();
		logger.debug("Condition locked!");
		
		cond.signal();
		logger.debug("Condition signal!");
		lock.unlock();
			
	}
	
	public void CloseWatcher(){
		try {
			
			logger.debug("close watcher ");
	
			
			Stat st = zk.exists(FilePath, false);
			if(st != null){
				List<String>ChildList = zk.getChildren(FilePath, this);
				
				for(String child : ChildList){
					logger.debug("child file is:" + child);
					zk.delete(FilePath + "/" + child, -1); 
				}
				
				zk.delete(FilePath, -1); 
			}
			if(zk != null){
				zk.close();
				zk = null;
			}
			
		} catch (InterruptedException | KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	

	public String getHosts() {
		return hosts;
	}

	public void setHosts(String hosts) {
		this.hosts = hosts;
	}

	public String getFilePath() {
		return FilePath;
	}

	public void setFilePath(String filePath) {
		FilePath = filePath;
	}
	
}
